#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
RocketMQ 消费者演示
演示如何从 RocketMQ 消费消息
"""

import time
import json
import logging
from rocketmq.client import PushConsumer, ConsumeStatus, ReceivedMessage
from config import ROCKETMQ_CONFIG, MESSAGE_CONFIG, LOGGING_CONFIG, DEMO_CONFIG, check_config, configure_rocketmq

# 配置日志
logging.basicConfig(
    level=getattr(logging, LOGGING_CONFIG["level"]), 
    format=LOGGING_CONFIG["format"]
)
logger = logging.getLogger(__name__)


class MessageHandler:
    """消息处理器"""
    
    def __init__(self):
        self.message_count = 0
        self.start_time = time.time()
    
    def __call__(self, msg: ReceivedMessage):
        """消息处理回调"""
        try:
            self.message_count += 1
            current_time = time.time()
            elapsed_time = current_time - self.start_time
            
            logger.info(f"=== 收到消息 #{self.message_count} (运行时间: {elapsed_time:.1f}s) ===")
            
            # 基本信息
            logger.info(f"  - Topic: {msg.topic}")
            logger.info(f"  - Tags: {msg.tags}")
            logger.info(f"  - Keys: {msg.keys}")
            logger.info(f"  - MsgId: {msg.id}")
            logger.info(f"  - QueueId: {getattr(msg, 'queue_id', 'N/A')}")
            
            # 兼容性处理：offset 可能不存在
            if hasattr(msg, 'offset'):
                logger.info(f"  - Offset: {msg.offset}")
            
            # 消息内容
            try:
                body = msg.body.decode(MESSAGE_CONFIG["encoding"])
                logger.info(f"  - 消息内容: {body}")
                
                # 尝试解析 JSON
                try:
                    data = json.loads(body)
                    if isinstance(data, dict):
                        logger.info(f"  - 消息ID: {data.get('id', 'N/A')}")
                        logger.info(f"  - 消息类型: {data.get('data', {}).get('type', 'N/A')}")
                        logger.info(f"  - 时间戳: {data.get('timestamp', 'N/A')}")
                except json.JSONDecodeError:
                    pass  # 不是 JSON 格式，忽略
                    
            except Exception as e:
                logger.warning(f"  - 消息内容解码失败: {e}")
                logger.info(f"  - 原始内容: {msg.body}")
            
            logger.info("=== 消息处理完成 ===")
            
            return ConsumeStatus.CONSUME_SUCCESS
            
        except Exception as e:
            logger.error(f"消息处理失败: {e}")
            return ConsumeStatus.RECONSUME_LATER


class RocketMQConsumer:
    """RocketMQ 消费者类"""
    
    def __init__(self):
        self.consumer = None
        self.message_handler = MessageHandler()
        self.namesrv_addr = ROCKETMQ_CONFIG["namesrv_addr"]
        self.access_key = ROCKETMQ_CONFIG["access_key"]
        self.secret_key = ROCKETMQ_CONFIG["secret_key"]
        self.group_name = ROCKETMQ_CONFIG["consumer_group"]
        self.topic = ROCKETMQ_CONFIG["default_topic"]
        self.message_tags = ROCKETMQ_CONFIG["consumer_config"]["message_tags"]
        self.orderly = ROCKETMQ_CONFIG["consumer_config"]["orderly"]
    
    def start(self):
        """启动消费者"""
        try:
            logger.info("=== 启动 RocketMQ 消费者 ===")
            logger.info(f"NameServer: {self.namesrv_addr}")
            logger.info(f"消费者组: {self.group_name}")
            logger.info(f"主题: {self.topic}")
            logger.info(f"消息标签: {self.message_tags}")
            logger.info(f"有序消费: {self.orderly}")
            
            # 创建消费者
            self.consumer = PushConsumer(
                self.group_name,
                orderly=self.orderly
            )
            self.consumer.set_name_server_address(self.namesrv_addr)
            self.consumer.set_session_credentials(self.access_key, self.secret_key, "ALIYUN")
            
            # 订阅主题
            self.consumer.subscribe(self.topic, self.message_handler, self.message_tags)
            
            # 启动消费者
            self.consumer.start()
            logger.info("✓ 消费者启动成功")
            logger.info("正在等待消息...")
            return True
            
        except Exception as e:
            logger.error(f"✗ 消费者启动失败: {e}")
            return False
    
    def run(self, timeout=None):
        """运行消费者"""
        if timeout is None:
            timeout = DEMO_CONFIG["consume_timeout"]
        
        try:
            start_time = time.time()
            
            while time.time() - start_time < timeout:
                time.sleep(1)
                
                # 每5秒打印一次状态
                elapsed = time.time() - start_time
                if int(elapsed) % 5 == 0 and elapsed > 0:
                    logger.info(f"消费者运行中... 已收到 {self.message_handler.message_count} 条消息")
            
            logger.info(f"消费者运行完成，共收到 {self.message_handler.message_count} 条消息")
            
        except KeyboardInterrupt:
            logger.info("用户中断消费者")
        except Exception as e:
            logger.error(f"消费者运行错误: {e}")
    
    def shutdown(self):
        """关闭消费者"""
        if self.consumer:
            self.consumer.shutdown()
            logger.info("消费者已关闭")


def check_configuration():
    """检查配置"""
    missing_fields = check_config()
    if missing_fields:
        print(f"❌ 配置不完整，缺少: {', '.join(missing_fields)}")
        print("请先配置 RocketMQ 连接信息")
        
        if configure_rocketmq():
            print("✓ 配置完成，可以继续运行演示")
            return True
        else:
            print("❌ 配置失败，请重新运行程序")
            return False
    return True


def main():
    """主函数"""
    logger.info("开始 RocketMQ 消费者演示...")
    
    # 检查配置
    if not check_configuration():
        return
    
    # 创建消费者
    consumer = RocketMQConsumer()
    
    try:
        # 启动消费者
        if not consumer.start():
            return
        
        # 运行消费者
        consumer.run()
        
        # 总结
        logger.info("=== 演示总结 ===")
        logger.info(f"收到消息数量: {consumer.message_handler.message_count}")
        logger.info(f"运行时间: {time.time() - consumer.message_handler.start_time:.1f} 秒")
        
        if consumer.message_handler.message_count > 0:
            logger.info("✓ 消费者演示成功完成")
        else:
            logger.warning("⚠ 未收到任何消息，请检查是否有消息发送到该主题")
            
    except Exception as e:
        logger.error(f"演示过程中发生错误: {e}")
    finally:
        consumer.shutdown()


if __name__ == "__main__":
    main() 